/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package dsmsforkjoin;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.ServerSocket;
import java.net.Socket;
import tuplegenerator.MeterUser;
import tuplegenerator.Tuple;

/**
 *
 *
 */
public class InputHandlerUserMeters<T extends Tuple> extends AbstractInputHandler<T> {

    private ServerSocket server;
    private Socket connection;
    private BufferedReader reader;

    public InputHandlerUserMeters(RunMe rm, FJoin<T> fjoin) {
        super(rm, fjoin);
    }
 
    @SuppressWarnings("unchecked")
	@Override
    public void run() {
        long cnt = 0;
        
        try {
            server = new ServerSocket(5556, 100);
            
            this.waitConnection();
            this.getInputRoute();
                
            if (reader == null) {
                this.closeAllConnections();
                return;
            }
            
            String data = null;
            int tp = 0;
            while ((data = reader.readLine()) != null) {
                MeterUser mu = MeterUser.parse(data);
                if (mu == null) {
                    System.out.println("Ignored a tuple!");
                } else {
                    //System.out.println(t.toString());

                    tp++;
                    mu.setIndex(tp);
                    flist.addTuple(tp - 1, (T) mu);
                    if (tp == RunMe.MAXIMUM_FORKS) {
                        tp = 0;
                    }
                    
                    if (++cnt >= RunMe.TOTAL_TUPLES) break;
                }
                
            }
            
            long ss = System.currentTimeMillis();
            
            ForkListSimpleQueue<T> fsq = (ForkListSimpleQueue<T>)flist;
            for (int i=0;i<fsq.listCount();i++) {
                fsq.getQueue(i).sort();
                System.out.println("Queue Users#" + (i+1) + ": " + fsq.getQueue(i).getSize());
            }
            

            
            long ee = System.currentTimeMillis();
            
            System.out.println("Time = " + (ee-ss) + " ms");
            System.out.println("Tuple recieve completed!\nApplication terminating.");
            
            this.runMe.reduce();
                        
        } catch (Exception ex) {
            ex.printStackTrace();
            
        } finally {
            this.closeAllConnections();
        }
         
    }
    
    private void waitConnection() throws IOException {
        System.out.println("Waiting for connection...");
        connection = server.accept();
        System.out.println("Connection established: " +
                connection.getInetAddress().getHostName());
    }
    
    private void getInputRoute() throws IOException {
        reader = new BufferedReader(new InputStreamReader(connection.getInputStream()));
    }
    private void closeAllConnections() {

        try {
             reader.close();
             connection.close();
        } catch (IOException e) {
            e.printStackTrace();
        }

    } // end of closeAllConnections method.

    
}
